package core.proxy;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.Date;

import org.apache.log4j.Logger;

import core.comunication.FileClient;
import core.comunication.MSGTDEnqueue;
import core.comunication.MSGTDEnqueueAck;
import core.descriptors.TaskDescriptor;
import core.enums.Errors;
import core.enums.TaskTYPE;
import core.manager.TQueueManager;
import core.util.COREServiceRegistry;
import core.util.SockUtil;
import core.worker.Address;

/**
 * Implementazione specifica del TQueueManager proxy per il P2P.
 * @see TQueueManagerProxy
 * @author Vincenzo Merlo, Scuderi Giovanni Luca, Paolo Giuffrida
 */
public final class TQueueManagerProxyP2P extends TQueueManagerProxy {

    private final Logger logger = COREServiceRegistry.getLogger(TQueueManagerProxyP2P.class.toString()); //istanza log4j

    /**
     * Metodo che non deve essere usato in quanto la dequeu non puo' essere effetuata remotamente ma solo localmente
     * @return null sempre
     */
    @Override
    public TaskDescriptor dequeue() {
        logger.debug("dequeue");
        return null;
    }

    /**
     * Accodamento remoto del TaskDescriptor nel TQueueManager 
     * @param td
     * @return 1 se va bene, ErrorCode altrimenti
     * @see Errors
     */
    @Override
    public int enqueue(TaskDescriptor td) {
        DatagramPacket dp = null;
        DatagramSocket ds = null;
        //retrieve del link
        try {

            File file = new File(td.getLink());
            int error;
            if (file.isFile()) {
                if ((error = new FileClient(InetAddress.getByName(getAddr().getIP()), getAddr().getPort(), file).run()) < 0) {
                    return error;
                } else {
                    Errors.FileNotFound.code();
                }
            }

            
            logger.debug(td);
            Long currentTime = new Date().getTime();
            byte[] message2Send = marshalling(td, currentTime);
            byte[] response = new byte[SockUtil.maxSize];

            //Enqueue Taskdescriptor

            boolean fault = false;
            int nRetrive = 0;
            do {
                fault = false;
                try {
                    logger.debug("Inzio l'enqueue del td tramite socket UDP - ProxyP2P tdID: " + td.getID() + " nRetrive: " + nRetrive);
                    logger.debug("Creazione datagram di invio - Address " + getAddr());
                    dp = new DatagramPacket(
                            message2Send,
                            message2Send.length,
                            InetAddress.getByName(getAddr().getIP()),
                            getAddr().getPort());

                    ds = new DatagramSocket();
                    ds.send(dp);
                    ds.close();
                    ds = null;
                    //response from server udp
                    ds = new DatagramSocket(getAddr().getPort() + 1);
                    ds.setReceiveBufferSize(SockUtil.maxSize);
                    dp = new DatagramPacket(response, response.length);
                    ds.setSoTimeout(SockUtil.timeout * 2);
                    ds.receive(dp);

                    unMarshalling(dp.getData(), currentTime, td.getID());

                } catch (SocketTimeoutException e) {
                    logger.debug(this.getClass().getCanonicalName() + " " + e.getClass().getSimpleName()
                            + " wait ack from server");
                    fault = true;
                    nRetrive++;
                } catch (AckExcemption e) {
                    logger.debug(this.getClass().getCanonicalName() + " " + e.getClass().getSimpleName()
                            + " ack errato");
                    fault = true;
                    nRetrive++;
                } finally {

                    ds.close();
                    ds = null;
                    dp = null;

                    if (nRetrive == SockUtil.nRetrive) {
                        logger.debug("Enqeue annullato");
                        return Errors.EnqueueTD.code();
                    }


                    System.gc();
                }
            } while (fault);
        } catch (UnknownHostException e) {
            logger.error(e.getStackTrace());
            return Errors.UnknownHostException.code();
        } catch (SocketException e) {
            logger.error(e.getStackTrace());
            return Errors.SocketException.code();
        } catch (IOException e) {
            logger.error(e.getStackTrace());
            return Errors.IOException.code();
        } catch (Exception e) {
            logger.error(e.getStackTrace());
            return Errors.Exception.code();
        } finally {
            if (ds != null) {
                ds.close();
            }
        }

        return 1;
    }

    /**
     * Prepara il TaskDescriptor per la comunicazione, inserendo un timestamp per segnare il pacchetto, in maniera da identificare eventuali duplicati
     * @param td  the taskdescriptor da inviare
     * @return oggetto preparato per la trasmissione remota tramite un array i byte
     * @throws IOException 
     */
    private byte[] marshalling(TaskDescriptor td, Long currentTime) throws IOException {
        return marshallingHL(td, currentTime);
    }

    /**
     * Metodo usato per il marshaling, dell'oggetto TaskDescriptor.
     * @param td TaskDescriptor da trasmettere
     * @param currentTime timestamp per segnare il pacchetto
     * @return arraybyte pronto per la trasmissione
     * @throws IOException 
     */
    private byte[] marshallingHL(TaskDescriptor td, Long currentTime) throws IOException {
        ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
        ObjectOutputStream objectStream;
        objectStream = new ObjectOutputStream(byteOut);
        MSGTDEnqueue message = new MSGTDEnqueue(InetAddress.getByName(getAddr().getIP()), td, currentTime);
        objectStream.writeObject(message);
        return byteOut.toByteArray();
    }

    /**
     * Conversione pacchetto arrivato da remoto. Formato pacchetto.
     * Protocol startMessage|CurrentTime|Separator|Response|Separator|CRC|EndMessage
     * @param response
     * @return > 0 fine if response is "OK"
     * 		   < 0 error otherwise
     * @throws IOException 
     * @throws ClassNotFoundException 
     */
    private void unMarshalling(byte[] buffer, long timeCurrent, int tdId) throws AckExcemption, IOException, ClassNotFoundException {
        ByteArrayInputStream input = new ByteArrayInputStream(buffer/*, packet.getBuffer().length*/);
        ObjectInputStream oIn = new ObjectInputStream(input);
        Object ack = oIn.readObject();
        if (!(ack instanceof MSGTDEnqueueAck)) 
        	throw new AckExcemption();
        
        MSGTDEnqueueAck ackCheck = new MSGTDEnqueueAck();
        ackCheck.setCurrentTime(timeCurrent);
        ackCheck.setId(tdId);
        
        if (!(ackCheck.equals(ack))) {
        	logger.debug("Ho ricevuto l'ack dal server, lo controllo..errato");
            throw new AckExcemption();
        } else
        	logger.debug("Ho ricevuto l'ack dal server, lo controllo..ok");
    }

    /**
     * Metodo main di test
     * @param args
     */
    public static void main(String[] args) {
        /*TQueueManager manager = */
    	Address addr = new Address();
        addr.setIP("127.0.0.1");
        addr.setPort(SockUtil.portMessagesSender);
        
        new TQueueManager(addr);

        for (int i = 0; i < 40; i++) {
            System.out.println("lancio " + i);

            TQueueManagerProxyP2P p = new TQueueManagerProxyP2P();
            p.setAddr(addr);
            TaskDescriptor td = new TaskDescriptor();
            td.setID(22 + i);
            td.setCommand("foto.jpg");
            td.setLink("c:\\eserciziRMI.pdf");
            td.setType(TaskTYPE.Executable);

            p.enqueue(td);

        }
    }

    /**
     *
     * @return
     */
    @Override
    public String toString() {
        return TQueueManagerProxyP2P.class.getSimpleName();
    }
}

class AckExcemption extends Exception {

    /**
     *
     */
    private static final long serialVersionUID = 3902296017212185465L;

    AckExcemption(String message) {
        super(message);
    }

    AckExcemption() {
        super("Ho ricevuto un ack non valido, ritento la comunicazione");
    }
}
